#ifndef _M_SERVER_H__
#define _M_SERVER_H__

#include <iostream>
#include <vector>
#include <cassert>
#include <cstring>
#include <thread>
#include <memory>
#include <unistd.h>
#include <fcntl.h>
#include <mutex>
#include <functional>
#include <unordered_map>
#include <condition_variable>
#include <signal.h>
#include <sys/timerfd.h>
#include <sys/eventfd.h>
#include <sys/epoll.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include "../log/logger.hpp"
#include "../log/log.hpp"
#include "../log/err.hpp"

#define BUFFER_DEFAULT_SIZE 1024
class Buffer
{
public:
    Buffer()
    :_buffer(BUFFER_DEFAULT_SIZE),_reader_idx(0),_writer_idx(0)
    {}
    char* Begin() { return &*_buffer.begin(); }
    //获取当前写入起始位置
    char* WritePosition() { return Begin() + _writer_idx; }
    //获取当前读取起始位置
    char* ReadPosition() { return Begin() + _reader_idx; }
    //获取缓冲区末尾空闲空间
    uint64_t TailIdleSize() { return _buffer.size() - _writer_idx; }
    //获取缓冲区起始空闲空间
    uint64_t HeadIdleSize() { return _reader_idx; }
    //获取可读数据大小
    uint64_t ReadAblesSize() { return _writer_idx - _reader_idx; }
    //将读偏移向后移动
    void MoveReadOffset(uint64_t len)
    {
        if(len == 0) return;
        //必须保证有可读的数据大小
        assert(len <= ReadAblesSize());
        _reader_idx += len;
    }
    //将写偏移向后移动
    void MoveWriteOffset(uint64_t len)
    {
        assert(len <= TailIdleSize());
        _writer_idx += len;
    }
    //确保可写空间足够
    void EnsureWriteSpace(uint64_t len)
    {
        //末尾空闲空间
        if(TailIdleSize() >= len) 
            return;
        //再加上前言空闲空间
        if(len <= TailIdleSize() + HeadIdleSize())
        {
            //把数据移到起始位置
            uint64_t rsz = ReadAblesSize();
            //std::copy(start, end, std::back_inserter(container));
            std::copy(ReadPosition(),ReadPosition() + rsz,Begin());
            _reader_idx = 0;
            _writer_idx = rsz;
        }
        else//扩容
        {
            DBG_LOG("RESIZE %ld",_writer_idx + len);
            //logMessage(DEBUG,"RESIZE %ld",_writer_idx + len);
            _buffer.resize(_writer_idx + len);
        }
    }     
    //写入数据
    void Write(const void* data,uint64_t len)
    {
        //1.保证空间大小,拷贝数据
        if(len == 0) return;
        EnsureWriteSpace(len);
        const char* d = (const char*)data;
        std::copy(d,d+len,WritePosition());
    }
    //写入数据并后移
    void WriteandPush(const void* data,uint64_t len)
    {
        Write(data,len);
        MoveWriteOffset(len);
    }
    //写入string数据
    void WriteString(const std::string& data)
    {
        return Write(data.c_str(),data.size());
    }
    //写入string并后移
    void WriteStringAndPush(const std::string& data)
    {
        WriteString(data);
        MoveWriteOffset(data.size());
    }
    //写入buffer数据
    void WriteBuffer(Buffer& data)
    {
        return Write(data.ReadPosition(),data.ReadAblesSize());
    }
    //写入buffer并后移
    void WriteBufferAndPush(Buffer& data)
    {
        WriteBuffer(data);
        MoveWriteOffset(data.ReadAblesSize());
    }
    //读取数据
    void Read(void* buf,uint64_t len)
    {
        assert(len <= ReadAblesSize());
        std::copy(ReadPosition(),ReadPosition() + len,(char*)buf);
    }
    //读取数据并后移
    void ReadAndPop(void* buf,uint64_t len)
    {
        Read(buf,len);
        MoveReadOffset(len);
    }
    //读取string数据
    std::string ReadString(uint64_t len)
    {
        assert(len <= ReadAblesSize());
        std::string str;
        str.resize(len);
        Read(&str[0],len);
        return str;
    }
    //读取string并后移
    std::string ReadStringAndPop(uint64_t len)
    {
        assert(len <= ReadAblesSize());
        std::string str = ReadString(len);
        MoveReadOffset(len);
        return str;
    }
    //获取换行符
    char* FindCRLF()
    {
        char* res = (char*)memchr(ReadPosition(),'\n',ReadAblesSize());
        return res;
    }
    //获取一行数据
    std::string GetLine()
    {
        char* pos = FindCRLF();
        if(pos == NULL)
            return "";
        //+1 是为了把\n也取出来
        return ReadString(pos - ReadPosition() + 1);
    }
    //获取一行数据并后移
    std::string GetLineAndPop()
    {
        std::string str = GetLine();
        MoveReadOffset(str.size());
        return str;
    }
    //清空缓冲区
    void Clear()
    {
        _writer_idx = 0;
        _reader_idx = 0;
    }
private:
    std::vector<char> _buffer;
    uint64_t _reader_idx;
    uint64_t _writer_idx;
};

#define MAX_LISTEN 1024

class Socket
{
public:
    Socket():_socket_fd(-1)
    {}
    Socket(int sock):_socket_fd(sock)
    {}
    ~Socket() { Close(); }
    int Fd() { return _socket_fd; }
    //创建套接字
    bool Create()
    {
        // int socket(int domain, int type, int protocol)
        _socket_fd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);//默认最后一个参数给0,让其自己选择
        if(_socket_fd < 0)
        {
            ERR_LOG("create socket error!");
            return false;
        }
        return true;
    }
    //绑定套接字信息
    bool Bind(const std::string& ip,const uint64_t& port)
    {
        struct sockaddr_in addr;
        addr.sin_family = AF_INET;
        addr.sin_addr.s_addr = inet_addr(ip.c_str());
        addr.sin_port = htons(port);
        socklen_t len = sizeof(struct sockaddr_in);
        int ret = bind(_socket_fd,(struct sockaddr*)&addr,len);
        if(ret < 0)
        {
            ERR_LOG("socket bind error!");
            return false;
        }
        return true;
    }
    //进行监听
    bool Listen(int backlog = MAX_LISTEN)
    {
        int ret = listen(_socket_fd,backlog);
        if(ret < 0)
        {
            ERR_LOG("socket listen error!");
            return false;
        }
        return true;
    }
    //发起连接
    bool Connect(const std::string& ip,const uint64_t& port)
    {
        struct sockaddr_in addr;
        addr.sin_family = AF_INET;
        addr.sin_addr.s_addr = inet_addr(ip.c_str());
        addr.sin_port = htons(port);
        socklen_t len = sizeof(struct sockaddr_in);
        int ret = connect(_socket_fd,(struct sockaddr*)&addr,len);
        if(ret < 0)
        {
            ERR_LOG("socket connect error!");
            return false;
        }
        return true;
    }
    //获取新连接
    int Accept()
    {
        int fd = accept(_socket_fd,NULL,NULL);
        if(fd < 0)
        {
            ERR_LOG("socket accept error!");
            return -1;
        }
        return fd;
    }
    //接收数据
    ssize_t Recv(void* buf,size_t len,int flag = 0)
    {
        ssize_t ret = recv(_socket_fd,buf,len,flag);
        if(ret <= 0)
        {
            //EAGAIN 当前socket的接收缓冲区中没有数据了，在非阻塞的情况下才会有这个错误
            //EINTR  表示当前socket的阻塞等待，被信号打断了
            if(errno == EAGAIN || errno == EINTR){
                return 0;
            }
            ERR_LOG("socket recv error!");
            return -1;
        }
        DBG_LOG("%s---%d",buf,ret);
        //返回实际接收的长度
        return ret;
    }
    ssize_t RecvNonblack(void* buf,size_t len)
    {
        return Recv(buf,len,MSG_DONTWAIT);
    }
    //发送数据
    ssize_t Send(const void* buf,size_t len,int flag = 0)
    {
        ssize_t ret = send(_socket_fd,buf,len,flag);
        if(ret <= 0)
        {
            if(errno == EAGAIN || errno == EINTR)
                return 0;
            ERR_LOG("socket send error!");
            return -1;
        }
        DBG_LOG("%s---%d",buf,ret);
        //返回实际发送的长度
        return ret;
    }
    ssize_t SendNonblack(const void* buf,size_t len)
    {
        if(len == 0) return 0;
        return Send(buf,len,MSG_DONTWAIT);
    }
    //关闭套接字
    void Close()
    {
        if(_socket_fd != -1)
        {
            close(_socket_fd);
            _socket_fd = -1;
        }
            
    }
    //创建一个服务器监听连接
    bool CreateServer(uint16_t port, const std::string &ip = "0.0.0.0", bool block_flag = false) 
    {
        //1. 创建套接字，2. 绑定地址，3. 开始监听，4. 设置⾮阻塞， 5. 启动地址重⽤
        if (Create() == false) 
            return false;
        if (block_flag) Nonblack();
        if (Bind(ip, port) == false) 
            return false;
        if (Listen() == false) 
            return false;
        ReuseAddress();
        return true;
 }
    //创建一个客户端请求连接
    bool CreateClient(const std::string& ip,uint16_t port)
    {
        if(Create() == false) 
            return false;
        if(Connect(ip,port) == false) 
            return false;
        DBG_LOG("create client success");
        return true;
    }
    //设置非阻塞
    void Nonblack()
    {
        int flag = fcntl(_socket_fd,F_GETFL,0);
        fcntl(_socket_fd,F_SETFL,flag | O_NONBLOCK);
    }
    //开启端口复用
    void ReuseAddress()
    {
        // int setsockopt(int fd, int leve, int optname, void *val, int vallen)
        int val = 1;
        setsockopt(_socket_fd,SOL_SOCKET,SO_REUSEADDR,(void*)&val,sizeof(int));
        val = 1;
        setsockopt(_socket_fd,SOL_SOCKET,SO_REUSEPORT,(void*)&val,sizeof(int));
    }
private:
    int _socket_fd;
};

using EventCallBack = std::function<void()>;
class Poller;
class EventLoop;
class Channel
{
public:
    Channel(int fd,EventLoop* loop):_fd(fd),_events(0),_revents(0),_loop(loop) {}
    int Fd() { return _fd; }
    //获取事件
    uint32_t Events() {return _events;}
    void SetREvent(uint32_t events) { _revents = events; }
    //设置读事件回调
    void SetReadCallback(const EventCallBack &cb) {_read_callback = cb;} 
    //设置写事件回调
    void SetWriteCallback(const EventCallBack &cb) {_write_callback = cb;}
    //设置错误事件回调
    void SetErrorCallback(const EventCallBack &cb) {_error_callback = cb;}
    //设置关闭事件回调
    void SetCloseCallback(const EventCallBack &cb) {_close_callback = cb;}
    //设置所有事件回调
    void SetEventCallback(const EventCallBack &cb) {_event_callback = cb;}
    //当前是否监控可读
    bool ReadAble() { return (_events & EPOLLIN); }
    //当前是否监控可写
    bool WriteAble() { return (_events & EPOLLOUT); }
    //设置可读
    void EnableRead() { _events |= EPOLLIN; Update(); }
    //设置可写
    void EnableWrite() { _events |= EPOLLOUT; Update();}
    //删除可读
    void DisableRead() { _events &= ~EPOLLIN; Update(); }
    //删除可写
    void DisableWrite() { _events &= ~EPOLLOUT; Update(); } 
    //删除所有
    void DisableAll() { _events = 0; Update(); }
    //移除事件
    void Remove();
    //事件的新增/增信
    void Update();
    //事件处理,一旦连接触发了事件,就调用该函数,进行事件的回调
    void HandleEvent()
    {
        if((_revents & EPOLLIN) || (_revents & EPOLLRDHUP) || (_revents & EPOLLPRI))
        {
            if(_read_callback) _read_callback();
        }
        //
        if(_revents & EPOLLOUT)
        {
            if(_write_callback)
                _write_callback();
        }
        else if(_revents & EPOLLERR)
        {
            if(_error_callback)
                _error_callback();
        }
        else if(_revents & EPOLLHUP)
        {
            if(_close_callback)
                _close_callback();
        }
            
        //任意事件回调调用
        if(_event_callback)
            _event_callback();
    }

private:
    int _fd;
    EventLoop* _loop;
    uint32_t _events;//当前需要监控的事件
    uint32_t _revents;//当前连接触发的事件
    EventCallBack _read_callback; //可读事件回调
    EventCallBack _write_callback;//可写事件回调
    EventCallBack _close_callback;//关闭事件回调
    EventCallBack _error_callback;//错误事件回调
    EventCallBack _event_callback;//任意事件回调
};


#define MAX_EPOLLEVENTS 1024
class Channel;
class Poller
{
public:
    Poller()
    {
        _epfd = epoll_create(MAX_EPOLLEVENTS);
        if(_epfd < 0)
        {
            ERR_LOG("epoll_create error,code:%d,errstring:%s",errno,strerror(errno));
            abort();
        }
    }
    //添加/修改监控事件
    void UpdateEvent(Channel *channel)
    {
        bool ret = HasChannel(channel);
        if(ret == false)
        {
            //没有就添加
            _channels.insert(std::make_pair(channel->Fd(),channel));
            return Update(channel,EPOLL_CTL_ADD);
        }
        return Update(channel,EPOLL_CTL_MOD);
    }
    //移除监控事件
    void RemoveEvent(Channel *channel)
    {
        auto it = _channels.find(channel->Fd());
        if(it != _channels.end())
            _channels.erase(it);
        Update(channel,EPOLL_CTL_DEL);
    }
    //开始监控，返回连接
    void Poll(std::vector<Channel*> *active)
    {
        //对于timeout,后序修改
        // int epoll_wait(int epfd, struct epoll_event *evs, int maxevents, int timeout)
        int nfds = epoll_wait(_epfd,_evs,MAX_EPOLLEVENTS,-1);
        if(nfds < 0)
        {
            if(errno == EINTR)
                return;
            ERR_LOG("epoll_wait error:%s\n",strerror(errno));
            abort();
        }
        for(int i = 0; i < nfds; i++)
        {
           auto it = _channels.find(_evs[i].data.fd);
           assert(it != _channels.end());
           it->second->SetREvent(_evs[i].events);
           active->push_back(it->second);
        }
        return;
    }
private:
    //对epoll直接接口调用
    void Update(Channel *channel,int op)
    {
        //int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);
        int fd = channel->Fd();
        struct epoll_event ev;
        ev.data.fd = fd;
        ev.events = channel->Events();
        int ret = epoll_ctl(_epfd,op,fd,&ev);
        if(ret < 0)
        {
            ERR_LOG("epoll_ctl error!");
            //abort();//直接退出程序
        }
        return;
    }
    //判断一个channel是否添加了事件监控
    bool HasChannel(Channel *channel)
    {
        auto it = _channels.find(channel->Fd());
        if(it == _channels.end())
            return false;
        return true;
    }
private:
    int _epfd;
    struct epoll_event _evs[MAX_EPOLLEVENTS];
    std::unordered_map<int,Channel *> _channels;
};

using TaskFunc = std::function<void()>;
using ReleaseFunc = std::function<void()>;
class TimerTask
{
public:
    TimerTask(uint64_t id,uint32_t time,const TaskFunc& cb)
    :_id(id),_timeout(time),_task_cb(cb),_canceled(false)
    {}
    ~TimerTask()
    {
        //先判断该任务是否被取消了,没被取消就执行相应的任务
        if(_canceled == false)
            _task_cb();
        _release();
    }
    //设置定时任务状态
    void set_canceled() { _canceled = true; }
    //设置定时任务的信息
    void set_release(const ReleaseFunc& func) { _release = func; }
    //获取定时时间
    uint32_t get_delay_timer() { return _timeout; }
private:
    uint64_t _id; //定时任务id
    uint32_t _timeout;//定时时间
    TaskFunc _task_cb;//定时任务的回调函数
    ReleaseFunc _release;//删除该定时任务信息
    bool _canceled;//取消定时任务(默认为false -- 不取消, true -- 取消)
};

using WeakTask = std::weak_ptr<TimerTask>;//保存轮子中的任务信息
using PtrTask = std::shared_ptr<TimerTask>;//对应轮子中的每个任务
class TimerWheel
{
public:
    TimerWheel(EventLoop* loop)
    :_tick(0),_capacity(60),_wheel(_capacity)
    ,_loop(loop),_timerfd(CreateTimerFd())
    ,_timer_channel(new Channel(_timerfd,_loop))
    {
        _timer_channel->SetReadCallback(std::bind(&TimerWheel::OnTime,this));
        _timer_channel->EnableRead(); 
    }
    ~TimerWheel()
    {}
    /**
     * 定时器中有个_timers成员,定时器信息的操作可能是在多线程中进行,因此需要考虑线程安全问题
     * 但是加锁就会降低效率,所以就把对定期执行的所有操作放到一个线程中执行
    */
    //新增定时任务 
    void TimerAdd(uint64_t id,uint32_t time,const TaskFunc& cb);
    //刷新定时任务
    void TimerRefresh(uint64_t id);
    //移除定时任务
    void TimerCancel(uint64_t id);

    //是否存在某个定时任务(有线程安全,只能在对应EventLoop中调用)
    bool HasTimer(uint64_t id)
    {
        auto it = _timers.find(id);
        if(it == _timers.end())
            return false;
        return true;
    }
private:
    //根据id判断该任务是否存在
    void RemoveTimerTask(uint64_t id)
    {
        auto it = _timers.find(id);
        if(it != _timers.end())
            _timers.erase(it);
    }
    //创建timerfd,也就是定时时间函数相关设置
    static int CreateTimerFd()
    {
        /*int timerfd_settime(int fd, int flags,
                           const struct itimerspec *new_value,
                           struct itimerspec *old_value);*/
        int timerfd = timerfd_create(CLOCK_MONOTONIC,0);
        if(timerfd < 0)
        {
            ERR_LOG("create timerfd error!");
            //exit(TIMERFD_ERR);
            abort();
        }
        struct itimerspec itime;
        itime.it_value.tv_sec = 1;
        itime.it_value.tv_nsec = 0;//第一次超时时间为1s后
        itime.it_interval.tv_sec = 1;
        itime.it_interval.tv_nsec = 0;//第一次超时后，每次超时的间隔时间
        timerfd_settime(timerfd,0,&itime,nullptr);
        return timerfd;
    }
    //该函数应该一秒执行一次，相当于秒针向后走了一步
    void RunTimerTask()
    {
        _tick = (_tick + 1) % _capacity;
        _wheel[_tick].clear();//清除当前位置的信息,让该tick的shard_ptr释放掉,去执行析构函数
    }
    //获取定时任务id
    int ReadTimerFd()
    {
        uint64_t times;
        // int ret = read(_timerfd,&times,sizeof(times));
        int ret = read(_timerfd,&times,8);
        if(ret < 0)
        {
            ERR_LOG("read timerfd error");
            //exit(TIMERFD_ERR);
            abort();
        }
        return times;
    }
    //执行定时任务
    void OnTime()
    {
        int times = ReadTimerFd();
        for(int i = 0; i < times; i++){
            RunTimerTask();
        }    
    }
    //新增定时任务
    void TimerAddInloop(uint64_t id,uint32_t time,const TaskFunc& cb)
    {
        PtrTask pt(new TimerTask(id,time,cb));
        pt->set_release(std::bind(&TimerWheel::RemoveTimerTask,this,id));
        int pos = (_tick + time) % _capacity;
        _wheel[pos].push_back(pt);
        _timers[id] = WeakTask(pt);
    }
    //(刷新/延时)定时任务
    void TimerRefreshInLoop(uint64_t id)
    {
        auto it = _timers.find(id);
        if(it == _timers.end())
            return;
        PtrTask pt = it->second.lock();//lock获取weak_ptr管理对象对应的shared_ptr
        uint32_t delay = pt->get_delay_timer();
        int pos = (_tick + delay) % _capacity;
        _wheel[pos].push_back(pt);
    }
    //移除定时任务
    void TimerCancelInLoop(uint64_t id)
    {
        auto it = _timers.find(id);
        if(it == _timers.end())
            return;
        PtrTask pt = it->second.lock();//lock获取weak_ptr管理对象对应的shared_ptr
        if(pt) pt->set_canceled();
    }
private:
    int _tick;//当前位置
    int _capacity;//轮子最大延时时间
    std::vector<std::vector<PtrTask>> _wheel;//二维数组存放的是每个tick下存放的定时任务信息
    std::unordered_map<uint64_t,WeakTask> _timers;//通过任务id获取到任务
    EventLoop* _loop;//任务池
    int _timerfd;//定时器描述符--可读事件回调就是读取计数器,执行定时任务
    std::unique_ptr<Channel> _timer_channel;
};

using Functor = std::function<void()>;
class Channel;
class EventLoop
{
public:
    EventLoop()
    :_thread_id(std::this_thread::get_id())
    ,_event_fd(CreateEventFd())
    ,_event_channel(new Channel(_event_fd,this))
    ,_timer_wheel(this)
    {
        //给eventfd添加可读事件回调函数,读取eventfd事件通知次数
        _event_channel->SetReadCallback(std::bind(&EventLoop::ReadEventFd,this));
        //启动eventfd的读事件监控
        _event_channel->EnableRead();
    }
    //判断将要执行的任务是否处于当前线程,如果存在则执行,不是则压入队列
    void RunInLoop(const Functor& cb)
    {
        if(IsInLoop())
            return cb();
        return QueueLoop(cb);
    }
    void AssertInLoop()
    {
        assert(_thread_id == std::this_thread::get_id());
    }
    //将任务压入任务池
    void QueueLoop(const Functor& cb)
    {
        {
            std::unique_lock<std::mutex> _lock(_mutex);
            _tasks.push_back(cb);  
        }
        //需要唤醒有可能因为没有事件就绪,而导致epoll阻塞
        WeakUpEventFd();
    }
    //用于判断当前线程是否是EventLoop对应的线程
    bool IsInLoop() { return (_thread_id == std::this_thread::get_id()); }
    //增加/修改描述符事件监控
    void UpdateEvent(Channel* channel) { return _poller.UpdateEvent(channel); }
    //移除描述符事件监控
    void RemoveEvent(Channel* channel) { return _poller.RemoveEvent(channel); }

    /**
     * 对定时器操作函数
     * 增/修/删
    */
    void TimerAdd(uint64_t id,uint32_t delay,const TaskFunc &cb)
    {
        return _timer_wheel.TimerAdd(id,delay,cb);
    }
    void TimerRefresh(uint64_t id)
    {
        return _timer_wheel.TimerRefresh(id);
    }
    void TimerCancel(uint64_t id)
    {
        return _timer_wheel.TimerCancel(id);
    }
    //判断是否还有任务
    bool HasTimer(uint64_t id)
    {
        return _timer_wheel.HasTimer(id);
    }
    //开始--事件监控,就绪事件处理,执行任务
    void Start()
    {
        while(1)
        {
            //1.开始监控
            std::vector<Channel *> actives;
            _poller.Poll(&actives);
            //2.事件处理
            for(auto& e : actives)
            {
                e->HandleEvent();
            }
            //3.执行任务
            RunAllTask();
        }
    }
public:
    //执行任务池中的所有任务
    void RunAllTask()
    {
        std::vector<Functor> functor;
        {
            std::unique_lock<std::mutex> _lock(_mutex);
            _tasks.swap(functor);
        }
        for(auto &f: functor) { f(); }
        return ;
    }
    //创建eventfd信息
    static int CreateEventFd()
    {
        int efd = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK);
        if(efd < 0)
        {
            ERR_LOG("create eventfd error");
            abort();
        }
        return efd;
    }
    //获取eventfd
    void ReadEventFd()
    {
        uint64_t res = 0;
        int ret = read(_event_fd,&res,sizeof(res));
        if(ret < 0)
        {
            if(errno == EINTR || errno == EAGAIN) { 
                return; }
            ERR_LOG("read eventfd error!");
            abort();
        }
        return ;
    }
    //唤醒所有可能因为没有事件就绪而导致epoll的阻塞;
    void WeakUpEventFd()
    {
        uint64_t val = 1;
        int ret = write(_event_fd,&val,sizeof(val));
        if(ret < 0)
        {
            // if(errno == EINTR || errno == EAGAIN) { return; }
            if(errno == EINTR ) { 
                return; }
            ERR_LOG("read eventfd error!");
            abort();
        }
        return ;
    }
private:
    std::thread::id _thread_id;//线程ID
    int _event_fd;//eventfd唤醒IO事件监控id
    //Channel *_event_channel;
    std::unique_ptr<Channel> _event_channel;
    Poller _poller;//是所有描述符的事件监控
    std::vector<Functor> _tasks;//任务池
    TimerWheel _timer_wheel;
    std::mutex _mutex;//锁
};

class LoopThread
{
public:
    /*创建线程，设定线程入口函数*/
    LoopThread()
    :_loop(NULL), _thread(std::thread(&LoopThread::ThreadEntry, this)) 
    {}
    /*返回当前线程关联的EventLoop对象指针*/
    EventLoop *GetLoop() 
    {
        EventLoop *loop = NULL;
        {
            std::unique_lock<std::mutex> lock(_mutex);//加锁
            _cond.wait(lock, [&](){ return _loop != NULL; });//loop为NULL就一直阻塞
            loop = _loop;
        }
        return loop;
    }
private:
    /*实例化 EventLoop 对象，唤醒_cond上有可能阻塞的线程，并且开始运行EventLoop模块的功能*/
    void ThreadEntry() {
        EventLoop loopp;
        {
            std::unique_lock<std::mutex> lock(_mutex);//加锁
            _loop = &loopp;
            _cond.notify_all();//唤醒可能阻塞的线程
        }
        loopp.Start();
        }
private:
    /*用于实现_loop获取的同步关系，避免线程创建了，但是_loop还没有实例化之前去获取_loop*/
    std::mutex _mutex;          // 互斥锁
    std::condition_variable _cond;   // 条件变量
    EventLoop *_loop;       // EventLoop指针变量，这个对象需要在线程内实例化
    std::thread _thread;    // EventLoop对应的线程
};

class LoopThreadPool {
    private:
        int _thread_count;
        int _next_idx;
        EventLoop *_baseloop;
        std::vector<LoopThread*> _threads;
        std::vector<EventLoop *> _loops;
    public:
        LoopThreadPool(EventLoop *baseloop):_thread_count(0), _next_idx(0), _baseloop(baseloop) {}
        void SetThreadCount(int count) { _thread_count = count; }
        void Create() {
            if (_thread_count > 0) {
                _threads.resize(_thread_count);
                _loops.resize(_thread_count);
                for (int i = 0; i < _thread_count; i++) {
                    _threads[i] = new LoopThread();
                    _loops[i] = _threads[i]->GetLoop();
                }
            }
            return ;
        }
        EventLoop *NextLoop() {
            if (_thread_count == 0) {
                return _baseloop;
            }
            _next_idx = (_next_idx + 1) % _thread_count;
            return _loops[_next_idx];
        }
};


class Any{
    private:
        class holder {
            public:
                virtual ~holder() {}
                virtual const std::type_info& type() = 0;
                virtual holder *clone() = 0;
        };
        template<class T>
        class placeholder: public holder {
            public:
                placeholder(const T &val): _val(val) {}
                // 获取子类对象保存的数据类型
                virtual const std::type_info& type() { return typeid(T); }
                // 针对当前的对象自身，克隆出一个新的子类对象
                virtual holder *clone() { return new placeholder(_val); }
            public:
                T _val;
        };
        holder *_content;
    public:
        Any():_content(NULL) {}
        template<class T>
        Any(const T &val):_content(new placeholder<T>(val)) {}
        Any(const Any &other):_content(other._content ? other._content->clone() : NULL) {}
        ~Any() { delete _content; }

        Any &swap(Any &other) {
            std::swap(_content, other._content);
            return *this;
        }

        // 返回子类对象保存的数据的指针
        template<class T>
        T *get() {
            //想要获取的数据类型，必须和保存的数据类型一致
            assert(typeid(T) == _content->type());
            return &((placeholder<T>*)_content)->_val;
        }
        //赋值运算符的重载函数
        template<class T>
        Any& operator=(const T &val) {
            //为val构造一个临时的通用容器，然后与当前容器自身进行指针交换，临时对象释放的时候，原先保存的数据也就被释放
            Any(val).swap(*this);
            return *this;
        }
        Any& operator=(const Any &other) {
            /**
            * Any(val).swap(*this);
            * 表示的是val构造一个临时的通用容器
            * 然后于当前容器自身进行指针交换
            * 临时对象释放的时候，原先保存的数据也就被释放了
            */
            Any(other).swap(*this);
            return *this;
        }
};

//Connection
class Connection;
#define BUFFER_SIZE 65535  
/**
 * DISCONECTED-->连接关闭状态
 * CONNECTING-->连接启动状态
 * CONNECTED-->连接建立完成状态
 * DISCONNECTING-->待关闭状态
*/
typedef enum
{
    DISCONNECTED, 
    CONNECTING, 
    CONNECTED, 
    DISCONNECTING
}ConnStatu;
using PtrConnection = std::shared_ptr<Connection>;
class Connection : public std::enable_shared_from_this<Connection>
{
private:
    uint64_t _conn_id;//连接的唯一ID
    //对于_timer_id来讲是需要唯一的，所以这里直接使用_conn_id作为定时器ID
    int _sockfd;//套接字对应的id
    Socket _socket;//对于套接字的管理
    Channel _channel;//连接事件管理
    Buffer _in_buffer;//输入缓冲区
    Buffer _out_buffer;//输出缓冲区
    Any _context;//请求接收的上下文(协议控制)
    bool _enable_inactive_release;//控制非活跃连接销毁
    EventLoop* _loop;//连接对应的线程
    ConnStatu _statu;//当前连接所处状态
    /**
     * 回调函数的操作
     * 是让用户进行设置,供用户调用使用
     * ConnectedCallback _connected_callback -->连接管理回调
     * MessageCallback _message_callback-->信息管理回调
     * CloseCallback _close_callback-->挂壁连接回调
     * AnyEventCallback _event_callback-->任意事件回调
    */
    using ConnectedCallback = std::function<void(const PtrConnection&)>;
    using MessageCallback = std::function<void(const PtrConnection&, Buffer *)>;
    using ClosedCallback = std::function<void(const PtrConnection&)>;
    using AnyEventCallback = std::function<void(const PtrConnection&)>;
    ConnectedCallback _connected_callback;
    MessageCallback _message_callback;
    ClosedCallback _close_callback;
    AnyEventCallback _event_callback;
    /**
     * 组件内的连接关闭回调--组件内进行设置
     * 因为服务器组件会把所有的连接管理起来，一旦某个连接要关闭
     * 就应该从管理的地方移除掉其信息
    */
    ClosedCallback _server_close_callback;
public://weak_ptr
    // Connection(uint64_t conn_id,int sockfd,EventLoop* loop)
    // :_conn_id(conn_id),_sockfd(sockfd)
    // ,_enable_inactive_release(false)
    // ,_loop(loop)
    // ,_statu(CONNECTING)
    // ,_socket(_sockfd)
    // ,_channel(_sockfd,_loop)
    // {
    //     _channel.SetCloseCallback(std::bind(&Connection::HandleClose, this));
    //     _channel.SetEventCallback(std::bind(&Connection::HandleEvent, this));
    //     _channel.SetReadCallback(std::bind(&Connection::HandleRead, this));
    //     _channel.SetWriteCallback(std::bind(&Connection::HandleWrite, this));
    //     _channel.SetErrorCallback(std::bind(&Connection::HandleError, this));
    // }

    Connection(EventLoop *loop, uint64_t conn_id, int sockfd)
        :_conn_id(conn_id), _sockfd(sockfd)
        ,_enable_inactive_release(false),_loop(loop)
        ,_statu(CONNECTING),_socket(_sockfd),
        _channel(_sockfd,loop) 
        {
            _channel.SetCloseCallback(std::bind(&Connection::HandleClose, this));
            _channel.SetEventCallback(std::bind(&Connection::HandleEvent, this));
            _channel.SetReadCallback(std::bind(&Connection::HandleRead, this));
            _channel.SetWriteCallback(std::bind(&Connection::HandleWrite, this));
            _channel.SetErrorCallback(std::bind(&Connection::HandleError, this));
        }
    ~Connection() { DBG_LOG("release connection:%p",_conn_id); }
    //获取管理的文件描述符
    int SockFd() { return _sockfd; }
    //获取连接id
    int Id() { return _conn_id; }
    //是否处于连接完成状态
    bool Connected() { return (_statu == CONNECTED); }
    //设置上下文
    void SetContext(const Any& text) { _context = text; }
    //获取上下文
    Any* GetContext() { return &_context; }
    /* 设置相应的回调函数*/
    void SetConnectedCallback(const ConnectedCallback&cb) { _connected_callback = cb; }
    void SetMessageCallback(const MessageCallback&cb) { _message_callback = cb; }
    void SetClosedCallback(const ClosedCallback&cb) { _close_callback = cb; }
    void SetAnyEventCallback(const AnyEventCallback&cb) { _event_callback = cb; }
    void SetSrvClosedCallback(const ClosedCallback&cb) { _server_close_callback = cb; }
    //连接建立完成,进行channel回调设置,启动读事件监控,调用_connected_callback
    void Established()
    {
        _loop->RunInLoop(std::bind(&Connection::EstablishedInLoop,this));
    }
    //发送数据，将数据放到发送缓冲区，启动写事件监控
    void Send(const char* data,size_t size)
    {
        //外界传入的data，可能是个临时的空间，我们现在只是把发送操作压入了任务池，有可能并没有被立即执行
        //因此有可能执行的时候，data指向的空间有可能已经被释放了。
        Buffer buf;
        buf.WriteandPush(data, size);
        _loop->RunInLoop(std::bind(&Connection::SendInLoop, this, std::move(buf)));
        //_loop->RunInLoop(std::bind(&Connection::SendInLoop,this,data,size));
    }
    //提供给组件使用者的关闭接口--不是实际的关闭,需要判断有没有数据待处理
    void ShutDown()
    {
        _loop->RunInLoop(std::bind(&Connection::ShutDownInLoop,this));
    }
    //启动非活跃销毁，并定义多长时间无通信后就是非活跃连接,添加定时任务
    void EnableInactiveRelease(int sec)
    {
        _loop->RunInLoop(std::bind(&Connection::EnableInactiveReleaseInLoop,this,sec));
    }
    //取消非活跃销毁
    void CancelInactiveRelease()
    {
        _loop->RunInLoop(std::bind(&Connection::CancelInactiveReleaseInLoop,this));
    }
    //协议切换--重置上下文以及阶段性回调处理函数--也就是说该函数必须在EventLoop中执行
    //所以为了防备新事件触发后,处理的时候,以及切换任务单还没被执行就会使用原来的协议对该数据进行处理
    void Upgrade(const Any &connect,const ConnectedCallback& conn,
                 const MessageCallback& msg, const ClosedCallback& closed,
                 const AnyEventCallback& event)
    {
        _loop->AssertInLoop();
        _loop->RunInLoop(std::bind(&Connection::UpgradeInLoop,this,connect,conn,msg,closed,event));
    }
    //真正的释放
    void Release()
    {
        _loop->QueueLoop(std::bind(&Connection::ReleaseInLoop,this));
    }
private:
    /**channel相关事件回调函数*/
    //读事件触发后的回调函数
    void HandleRead()
    {
        //1.先从socket中读取数据,放到读缓冲区中
        // char buff[BUFFER_SIZE+1];
        char buff[65536];
        ssize_t ret = _socket.RecvNonblack(buff,65535);
        if(ret < 0)
        {
            //此时不能直接中断程序
            return ShutDownInLoop();
        }
        //这里等于0表示的是没有读取到数据，而不是断开连接了
        //此时把数据放到对应的输入缓冲区中,并把下标向后偏移
        _in_buffer.WriteandPush(buff,ret);
        //2.调用_message_callback进行业务的处理
        if(_in_buffer.ReadAblesSize() > 0)
        {
            /**
             * 从继承中的class Connection : public enable_shared_from_this<Connection>
             * enable_shared_from_this<Connection>会给我们提供一个weak_ptr
             * 我们从weak_ptr中获取对应shared_from_this()
             * 并能从中获得出一个shared_ptr
             * (shared_from_this()--从当前对象自身获取自身的shared_ptr管理对象)
            */
            return _message_callback(shared_from_this(),&_in_buffer);
        }
    }
    //写事件触发后的回调函数
    void HandleWrite()
    {
        //1.先从socket获取数据
        ssize_t ret = _socket.SendNonblack(_out_buffer.ReadPosition(),_out_buffer.ReadAblesSize());
        if(ret < 0)
        {
            if(_in_buffer.ReadAblesSize() > 0)
                 _message_callback(shared_from_this(),&_in_buffer);
            //return ReleaseInLoop();//直接进行释放
            return Release();
        }
        //将读偏移向后移动
        _out_buffer.MoveReadOffset(ret);
        //如果发送缓冲中没有数据
        if(_out_buffer.ReadAblesSize() == 0)
        {
            _channel.DisableWrite();//没有数据待发送了,直接关闭写事件监控
            if(_statu == DISCONNECTING){
                return Release();
                //return ReleaseInLoop();//直接进行释放
            }   
        }
        return;
    }

    //错误事件触发后的回调函数
    void HandleError()
    {
        return HandleClose();
    }
    //描述符触发关闭事件的回调函数
    void HandleClose()
    {
        //需要判断是否还有数据待处理
        if(_in_buffer.ReadAblesSize() > 0)
            _message_callback(shared_from_this(),&_in_buffer);
        //return ReleaseInLoop();
        return Release();
    }
    //任意事件触发后的回调函数
    void HandleEvent()
    {
        //1.刷新活跃的
        if(_enable_inactive_release == true)
            _loop->TimerRefresh(_conn_id);
        if(_event_callback)
            _event_callback(shared_from_this());
        //2.调用组件使用者的任意事件回调

    }
    //连接获取之后,所处的状态下进行的各种设置(Channel设置事件回调，启动读监控)
    void EstablishedInLoop()
    {
        //1.修改连接状态
        //当前的状态一定是构造函数时的初始状态CONNECTING
        assert(_statu == CONNECTING);
        //然后再将状态设置成连接建立完成状态
        _statu = CONNECTED;
        //2.启动读事件监控
        _channel.EnableRead();
        //3.调用回调函数
        if(_connected_callback)
            _connected_callback(shared_from_this());
    }
    //这个接口不是实际的发送接口,只是把数据放到缓冲区中,启动可写事件监控
    //实际的发送接口是HandleWrite
    void SendInLoop(Buffer& buff)
    {
        //1.判断连接状态
        if(_statu == DISCONNECTED)
            return ;
        //2.把数据放到发送缓冲区中,并启动写事件监控
        _out_buffer.WriteBufferAndPush(buff);
        if(_channel.WriteAble() == false)
            _channel.EnableWrite();
    }
    //不是真正的释放
    void ShutDownInLoop()
    {
        //1.修改连接状态为半关闭状态
        _statu = DISCONNECTING;
        //2.判断是否还有数据待处理
        if(_in_buffer.ReadAblesSize() > 0)
            if(_message_callback)
                _message_callback(shared_from_this(),&_in_buffer);
        if(_out_buffer.ReadAblesSize() > 0)
            if(_channel.WriteAble() == false)
                _channel.EnableWrite();
        if(_out_buffer.ReadAblesSize() == 0){
            //ReleaseInLoop();
            Release();
        }
    }
    //实际的释放接口
    void ReleaseInLoop()
    {
        //1.修改连接状态
        _statu = DISCONNECTED;
        //2.移除事件监控
        _channel.Remove();
        //3.关闭描述符
        _socket.Close();
        //4.检测定时器队列中还有任务,那么要将定时任务取消
        if(_loop->HasTimer(_conn_id))
            CancelInactiveReleaseInLoop();
        //5.调用关闭回调函数,避免先移除服务器管理信息造成Connection被释放,
        //  后序处理会造成野指针,因此要先调用用户的回调函数
        //  也就是先_close_callback()后_server_close_callback()
        if(_close_callback)
            _close_callback(shared_from_this());
        //6.移除服务器内部的连接信息
        if(_server_close_callback)
            _server_close_callback(shared_from_this());
    }
    //启动非活跃连接超时释放
    void EnableInactiveReleaseInLoop(int sec)
    {
        //1.将标志位设置为true
        _enable_inactive_release = true;
        //2.判断时候存在定时任务,存在则刷新延迟
        if(_loop->HasTimer(_conn_id))
            return _loop->TimerRefresh(_conn_id);
        //3.不存在则新增
        // _loop->TimerAdd(_conn_id,sec,std::bind(&Connection::ReleaseInLoop,this));
        _loop->TimerAdd(_conn_id,sec,std::bind(&Connection::Release,this));
    }
    //取消定时任务
    void CancelInactiveReleaseInLoop()
    {
        _enable_inactive_release = false;
        if(_loop->HasTimer(_conn_id)) 
            _loop->TimerCancel(_conn_id);
    }
    //协议切换
    void UpgradeInLoop(const Any &connect,const ConnectedCallback& conn,
                 const MessageCallback& msg, const ClosedCallback& closed,
                 const AnyEventCallback& event)
    {
        _context = connect;
        _connected_callback = conn;
        _message_callback = msg;
        _close_callback = closed;
        _event_callback = event;
    }
};

class Acceptor {
    private:
        Socket _socket;//用于创建监听套接字
        EventLoop *_loop; //用于对监听套接字进行事件监控
        Channel _channel; //用于对监听套接字进行事件管理

        using AcceptCallback = std::function<void(int)>;
        AcceptCallback _accept_callback;
    private:
        /*监听套接字的读事件回调处理函数---获取新连接，调用_accept_callback函数进行新连接处理*/
        void HandleRead() {
            int newfd = _socket.Accept();
            if (newfd < 0) {
                return ;
            }
            if (_accept_callback) _accept_callback(newfd);
        }
        int CreateServer(int port) {
            bool ret = _socket.CreateServer(port);
            assert(ret == true);
            return _socket.Fd();
        }
    public:
        /*不能将启动读事件监控，放到构造函数中，必须在设置回调函数后，再去启动*/
        /*否则有可能造成启动监控后，立即有事件，处理的时候，回调函数还没设置：新连接得不到处理，且资源泄漏*/
        Acceptor(EventLoop *loop, int port): _socket(CreateServer(port)), _loop(loop), 
            _channel(_socket.Fd(),loop) {
            _channel.SetReadCallback(std::bind(&Acceptor::HandleRead, this));
        }
        void SetAcceptCallback(const AcceptCallback &cb) { _accept_callback = cb; }
        void Listen() { _channel.EnableRead(); }
};

class TcpServer
{
private:
    uint64_t _next_id;//自增长ID
    int _port;//端口号
    int _timeout;//非活跃连接时间
    bool _enable_inactive_release;//启动超时销毁判断
    EventLoop _baseloop;//主线程的EventLoop,负责监听事件的处理
    Acceptor _acceptor;//监听套接字的管理对象
    LoopThreadPool _pool;//从属EventLoop线程池
    std::unordered_map<uint64_t,PtrConnection> _conns;//保存管理的连接对应的shared_ptr
    using ConnectedCallback = std::function<void(const PtrConnection&)>;
    using MessageCallback = std::function<void(const PtrConnection&,Buffer *)>;
    using CloseCallback = std::function<void(const PtrConnection&)>;
    using AnyEventCallback = std::function<void(const PtrConnection&)>;
    using Functor = std::function<void()>;
    ConnectedCallback _connected_callback;
    MessageCallback _message_callback;
    CloseCallback _close_callback;
    AnyEventCallback _event_callback;
private:
    //为新连接创建一个Connection进行管理
    void NewConnection(int fd)
    {
        DBG_LOG("new connection");
        _next_id++;
        //Connection(uint64_t conn_id,int sockfd,EventLoop* loop)
        PtrConnection conn(new Connection(_pool.NextLoop(),_next_id,fd));
        conn->SetMessageCallback(_message_callback);
        conn->SetClosedCallback(_close_callback);
        conn->SetAnyEventCallback(_event_callback);
        conn->SetConnectedCallback(_connected_callback);
        conn->SetSrvClosedCallback(std::bind(&TcpServer::RemoveConnection,this,std::placeholders::_1));
        if(_enable_inactive_release)
            conn->EnableInactiveRelease(_timeout);
        conn->Established();//初始化就绪
        _conns.insert(std::make_pair(_next_id,conn));
    }
    //添加一个定时任务
    void RunAfterInLoop(const Functor& task,int delay)
    {
        _next_id++;
        _baseloop.TimerAdd(_next_id,delay,task);
    }
    void RemoveConnection(const PtrConnection& conn)
    {
        _baseloop.RunInLoop(std::bind(&TcpServer::RemoveConnectionInLoop,this,conn));
    }
    //移除connection管理
    void RemoveConnectionInLoop(const PtrConnection& conn)
    {
        int fd = conn->Id();
        auto it = _conns.find(fd);
        if(it != _conns.end())
            _conns.erase(it);
    }
public:
    TcpServer(int port)
    :_next_id(0),_port(port),_enable_inactive_release(false)
    ,_acceptor(&_baseloop,_port)
    ,_pool(&_baseloop)
    {
        _acceptor.SetAcceptCallback(std::bind(&TcpServer::NewConnection,this,std::placeholders::_1));
        _acceptor.Listen();//将监听套接字挂到baseloop上开始事件的监听
    }
    //设置线程的数量
    void SetThreadCount(int count) { return _pool.SetThreadCount(count); }
    /* 设置相应的回调函数*/
    void SetConnectedCallback(const ConnectedCallback& cb) { _connected_callback = cb; }
    void SetMessageCallback(const MessageCallback& cb) { _message_callback = cb; }
    void SetClosedCallback(const CloseCallback& cb) { _close_callback = cb; }
    void SetAnyEventCallback(const AnyEventCallback& cb) { _event_callback = cb; }
    //设置启动非活跃连接超时销毁
    void EnableInactiveRelease(int timeout)
    {
        _timeout = timeout;
        _enable_inactive_release = true;
    }
    //添加一个定时任务
    void RunAfter(const Functor& task,int delay)
    {
        _baseloop.RunInLoop(std::bind(&TcpServer::RunAfterInLoop,this,task,delay));
    }
    //启动服务器
    void Start()
    {
        _pool.Create();//创建线程池中的从属线程
        _baseloop.Start();
    }
};

//移除事件监控
void Channel::Remove() { _loop->RemoveEvent(this); }
//事件的新增/更新
void Channel::Update() { _loop->UpdateEvent(this); }
//新增定时任务
void TimerWheel::TimerAdd(uint64_t id,uint32_t delay,const TaskFunc &cb)
{
    _loop->RunInLoop(std::bind(&TimerWheel::TimerAddInloop,this,id,delay,cb));
}
//刷新定时任务
void TimerWheel::TimerRefresh(uint64_t id)
{
    _loop->RunInLoop(std::bind(&TimerWheel::TimerRefreshInLoop,this,id));
}
//删除定时任务
void TimerWheel::TimerCancel(uint64_t id)
{ 
    _loop->RunInLoop(std::bind(&TimerWheel::TimerCancelInLoop,this,id));
}

class NetWork {
    public:
        NetWork() {
            DBG_LOG("SIGPIPE INIT");
            //logMessage(DEBUG,"SIGPIPE INIT");
            signal(SIGPIPE, SIG_IGN);
        }
};
static NetWork nw;

#endif